package cn.jx.asc.mqtt;


import cn.jx.asc.mqtt.handle.MessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.security.InvalidKeyException;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.CertificateException;

/**
 * @author z423
 * @2021-06-13-16:59
 * @effect    客户端连接MQ
 * @especial
 */
@Slf4j
@Component
public class MqClient {

    @Resource
    private MessageHandler handler;
    @Resource
    private mqttProperties mqpro;

    private final MqttConnectOptions options  = new MqttConnectOptions();
    private MqttClient client ;

    /**
     * 是否需要手动回复ack
     */
    private final boolean manualAcks = false;

    private String subTopic;

    public boolean getManualAcks(){

        return manualAcks;

    }

    /**
     * 消息推送完成
     * @param msg MqttMessage
     */
    public void messageArrivedComplete(MqttMessage msg){

        try {
            client.messageArrivedComplete(msg.getId(),msg.getQos());
        } catch (MqttException e) {
            log.error("send puback catch exception: {}", e.getMessage());
        }

    }

    /**
     * 连接 Mqtt 服务器
     * @return boolean
     */
    public synchronized boolean connect(){

        try {
            if(null == client) {
               client =  new MqttClient(mqpro.getServerUrl(), mqpro.getClientId(), new MemoryPersistence());
            }
            //获取连接配置
            client.setManualAcks(manualAcks);
            client.setCallback(new PushCallback(this, handler));
            resetOptions();
            try {
                client.connect(options);
            } catch (MqttException e) {
                log.error(e.getMessage());
            }
            subTopic = String.format("$sys/pb/consume/%s/%s/%s", mqpro.getObjName(), mqpro.getMqTopic(), mqpro.getMqSub());

            try {
                //订阅 topic $sys/pb/consume/$MQ_ID/$TOPIC/$SUB ，QoS必须大于0，否则订阅失败
                client.subscribe(subTopic, 1);
                log.info("sub success");
                return true;
            } catch (MqttException e) {
                log.error("sub failed: {}", e.getMessage());
            }
            return false;

        } catch (Exception e) {
            log.error(e.getMessage());
            return false;
        }

    }

    /**
     * 重新设置参数
     */
    private void resetOptions(){

        String expirationTime = System.currentTimeMillis() / 1000 + 100 * 24 * 60 * 60 + "";
        String password = null;
        try {
            //生成token
            password = Token.assembleToken(mqpro.getVersion(), "mqs/" + mqpro.getObjName(), expirationTime, mqpro.getSignatureMethod(), mqpro.getAccessKey());
        } catch (UnsupportedEncodingException | InvalidKeyException | NoSuchAlgorithmException e) {
            log.info(e.getMessage());
        }

        options.setCleanSession(true); //clean session 必须设置 true
        options.setUserName(mqpro.getObjName());
        assert password != null;
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(20);
        options.setKeepAliveInterval(30);
        options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
        InputStream caCrtFile;
        try{
            caCrtFile = this.getClass().getResource("/mqca/certificate.pem").openStream();
        } catch (IOException e){
            log.error(e.getMessage());
            return;
        }

        try {
            options.setSocketFactory(SslUtil.getSocketFactory(caCrtFile));
        } catch (NoSuchAlgorithmException | KeyStoreException | CertificateException
                | IOException | KeyManagementException e) {
            log.error(e.getMessage());
        }
    }

    /**
     * 掉线重连
     * @return boolean
     */
    public boolean reConnect() {

        if(null != client) {
            try {
                if(!client.isConnected()){ //订阅失败而导致重连是不需要重新连接
                    resetOptions();//防止token过期，需重新设置
                    client.connect(options);
                }
                client.subscribe(subTopic, 1);//订阅失败会抛异常
                log.info("reconncet and sub ok");
                return true;
            } catch (Exception e) {//订阅和连接失败都会进到此异常中
                log.info("reconnect failed: {}", e.getMessage());
                return false;
            }
        }else{
            return connect();
        }
    }
}
